package gupiaodown;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.nifi.annotation.behavior.ReadsAttribute;
import org.apache.nifi.annotation.behavior.ReadsAttributes;
import org.apache.nifi.annotation.behavior.WritesAttribute;
import org.apache.nifi.annotation.behavior.WritesAttributes;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.SeeAlso;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessorInitializationContext;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.OutputStreamCallback;
import org.apache.nifi.stream.io.ByteArrayOutputStream;

import hylanda.common.Transfer2JSON;
import net.minidev.json.JSONObject;

@Tags({" gentask gupiao price "})
@CapabilityDescription("gupiao price download ")
@SeeAlso({})
@ReadsAttributes({@ReadsAttribute(attribute="", description="")})
@WritesAttributes({@WritesAttribute(attribute="", description="")})

public class gendownGuPriceTaskProcessor extends AbstractProcessor {

	public gendownGuPriceTaskProcessor() {
		// TODO Auto-generated constructor stub
	}
	
	public static final Relationship MY_RELATIONSHIPOK = new Relationship.Builder()
            .name("gen success")
            .description("success relationship")
            .build();
	public static final Relationship MY_RELATIONSHIPNO = new Relationship.Builder()
            .name("gen failed")
            .description("failed relationship")
            .build();
	
	private List<PropertyDescriptor> descriptors;
    private Set<Relationship> relationships;
    
    
    @Override
    public Set<Relationship> getRelationships() {
        return this.relationships;
    }

    @Override
    public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
        return descriptors;
    }

 @Override
    protected void init(final ProcessorInitializationContext context) {
        final List<PropertyDescriptor> descriptors = new ArrayList<PropertyDescriptor>();
       
        this.descriptors = Collections.unmodifiableList(descriptors);
        final Set<Relationship> relationships = new HashSet<Relationship>();
        relationships.add(MY_RELATIONSHIPOK);
        relationships.add(MY_RELATIONSHIPNO);
        this.relationships = Collections.unmodifiableSet(relationships);
    }
 

	@Override
	public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
		// TODO Auto-generated method stub

		FlowFile flowFile = session.get();
	      if ( flowFile == null ) {
	      	
	      	System.out.println("flowfile is null");
	         return;
	      }
	      
	      //读取FlowFile
	        String content = "";
	        InputStream in = session.read(flowFile);
	        try {
				byte[] by = readStream(in);
				content = new String(by,"UTF-8");
				
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
				//return ;
			}
	        //格式{"name":"伊利股份","startdate":"2017-01-03","enddata":"2017-02-5"}//开始时间和结束时间可以不传或传空字符串
	        //单条或多条
	        DownAndAnalyzeGupiaoFromSina dotask = new DownAndAnalyzeGupiaoFromSina();
	        
	       List list =  dotask.genDownTask(content);
	       
	       for(int i = 0; i < list.size(); i++)
	       {
	    	   Map mapone = (Map) list.get(i);
	    	   String task = Transfer2JSON.TransferMap2JSON(mapone);
	    	  
	    	   if(i == 0)
	    	   {
	    		   flowFile= session.write(flowFile, new OutputStreamCallback(){
			    		@Override
			    		public void process(OutputStream out) throws IOException{
			    			out.write(task.getBytes("UTF-8"));
			    		}
			    	});
			     session.transfer(flowFile,MY_RELATIONSHIPOK);
	    	   }
	    	   else
	    	   {
	    		   flowFile = session.create();
	    		   flowFile= session.write(flowFile, new OutputStreamCallback(){
			    		@Override
			    		public void process(OutputStream out) throws IOException{
			    			out.write(task.getBytes("UTF-8"));
			    		}
			    	});
			     
			     session.transfer(flowFile,MY_RELATIONSHIPOK);
	    		   
	    	   }
	       }//for i

	}
	
	   public static byte[] readStream(InputStream inStream) throws Exception {  
	        ByteArrayOutputStream outSteam = new ByteArrayOutputStream();  
	        byte[] buffer = new byte[1024];  
	        int len = -1;  
	        while ((len = inStream.read(buffer)) != -1) {  
	            outSteam.write(buffer, 0, len);  
	        }  
	        outSteam.close();  
	        inStream.close();  
	        return outSteam.toByteArray();  
	    }  

}
